package com.yanggu.spark.core.demand.categoryTop10SessionTop10

import com.yanggu.spark.core.demand.categoryTop10.CategoryCountAccumulator
import org.apache.spark.{SparkConf, SparkContext}

//在需求一的基础上，增加每个品类用户session的点击统计
object Category_Top10_Session_Top10 {

  def main(args: Array[String]): Unit = {
    //1. 创建spark配置对象
    val sparkConf = new SparkConf().setAppName("Category_Top10_Session_Top10").setMaster("local[*]")

    //2. 创建spark上下文环境对象
    val sparkContext = new SparkContext(sparkConf)

    //3. 初始化累加器
    val categoryCountAccumulator = new CategoryCountAccumulator

    //4. 注册累加器
    sparkContext.register(categoryCountAccumulator, "categoryCountAccumulator")

    //5. 读取文件
    val fileRDD = sparkContext.textFile("input/user_visit_action.txt")
    fileRDD.cache()
    fileRDD
      //6. 循环遍历使用累加器进行聚合运算
      .foreach(line => {
        //使用模式匹配
        line.split("_") match {
          //计算点击次数
          case Array(_, _, _, _, _, "null", click_category_id, _, "null", "null", "null", "null", _) =>
            categoryCountAccumulator.add((click_category_id, "click"))
          //计算下单次数
          case Array(_, _, _, _, _, "null", "-1", "-1", order_category_ids, _, "null", "null", _) =>
            order_category_ids.split(",").map(id => categoryCountAccumulator.add((id, "order")))
          //计算支付次数
          case Array(_, _, _, _, _, "null", "-1", "-1", "null", "null", pay_category_ids, _, _) =>
            pay_category_ids.split(",").map(id => categoryCountAccumulator.add((id, "pay")))
          //如果是其他, 什么都不做
          case _ =>
        }
      })

    //累加器最后是将各个task中累加器的值调用merge方法进行合并, 最后将数据拉取到driver端。
    //因此如果累加器最后合并的数量过大, 可能会导致堆内存溢出。这点在使用时需要仔细考虑数据的规模
    //6. 获取累加器的值。并且排序, 得到top10
    val categoryIdList = categoryCountAccumulator
      .value
      .toList
      //这里使用tuple进行排序, 默认按照升序排序, 可以实现类似于sql中类似的功能。
      //例如先按照A升序, 然后按照B降序。如果要降序, 可以对值进行加上一个 "-" 号进行取反, 表示降序
      .sortBy {
        case (_, categoryCount) => (-categoryCount.clickCount, -categoryCount.orderCount, -categoryCount.payCount)
      }
      .map(_._1)
      .take(10)

    //7. 根据热门前10进行筛选过滤
    val map = fileRDD
      //7.1 过滤出点击事件、且分类为热门前10
      .filter(line => {
        val datas = line.split("_")
        !datas(6).equals("-1") && categoryIdList.contains(datas(6))
      })
      //7.2 进行结构的转换 => ((categoryId, sessionId), 1)
      .map { line =>
        val datas = line.split("_")
        ((datas(6), datas(2)), 1)
      }
      //7.3 进行累加求和 ((categoryId, sessionId), 1) => ((categoryId, sessionId), sum)
      .reduceByKey(_ + _)
      //7.4 进行结构的转换 ((categoryId, sessionId), sum) => (categoryId, (sessionId, sum))
      .map {
        case ((categoryId, sessionId), sum) =>
          (categoryId, (sessionId, sum))
      }
      //7.5 根据key进行分组 (categoryId, (sessionId, sum)) => (categoryId, Iterable[(sessionId, sum)])
      .groupByKey
      //7.6 key不变, 只对map的values进行操作。
      .mapValues {
        //得到List[sessionId, sum]
        _.toList
          //先根据sum进行降序排序。默认升序, 这里降序排序可以进行取反达到降序排序的目的
          .sortBy(-_._2)
          //获取前10个数量最多的sessionId
          .take(10)
      }
      //7.7 将数据拉取到driver端
      .collect
      //将数据转换成map。key-value的形式。这里是为了和原有的商品品类的热门排行保持一致
      .toMap

    //进行结构转换, 将原有的List[categoryId] => (categoryId, List[sessionId, sum])
    categoryIdList.map(categoryId => {
      //这里的Nil是为了保持泛型的统一
      (categoryId, map.getOrElse(categoryId, Nil))
      //这里进行循环, 遍历。使用了模式匹配, 同时过滤掉Nil
    }).foreach {
      case (categoryId, sessionAndSumList) if sessionAndSumList.nonEmpty =>
        println(s"分类id: $categoryId")
        sessionAndSumList.foreach {
          case (sessionId, sum) =>
            println(s"sessionId: $sessionId, 点击次数: $sum")
        }
        println
    }
  }

}
